// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//
#include <pollux/exec/addressable_non_null_value_list.h>
#include <pollux/exec/container_row_serde.h>

namespace kumo::pollux::aggregate::prestosql {
    ByteOutputStream AddressableNonNullValueList::initStream(
        HashStringAllocator *allocator) {
        ByteOutputStream stream(allocator);
        if (!firstHeader_) {
            // An array_agg or related begins with an allocation of 5 words and
            // 4 bytes for header. This is compact for small arrays (up to 5
            // bigints) and efficient if needs to be extended (stores 4 bigints
            // and a next pointer. This could be adaptive, with smaller initial
            // sizes for lots of small arrays.
            static constexpr int kInitialSize = 44;
            currentPosition_ = allocator->newWrite(stream, kInitialSize);
            firstHeader_ = currentPosition_.header;
        } else {
            allocator->extendWrite(currentPosition_, stream);
        }

        return stream;
    }

    AddressableNonNullValueList::Entry AddressableNonNullValueList::append(
        const DecodedVector &decoded,
        vector_size_t index,
        HashStringAllocator *allocator) {
        return append(*decoded.base(), decoded.index(index), allocator);
    }

    AddressableNonNullValueList::Entry AddressableNonNullValueList::append(
        const BaseVector &vector,
        vector_size_t index,
        HashStringAllocator *allocator) {
        auto stream = initStream(allocator);

        const auto hash = vector.hash_value_at(index);

        const auto originalSize = stream.size();

        // Write value.
        exec::ContainerRowSerdeOptions options{};
        exec::ContainerRowSerde::serialize(vector, index, stream, options);
        ++size_;

        auto startAndFinish = allocator->finishWrite(stream, 1024);
        currentPosition_ = startAndFinish.second;

        const auto writtenSize = stream.size() - originalSize;

        return {startAndFinish.first, writtenSize, hash};
    }

    HashStringAllocator::Position AddressableNonNullValueList::appendSerialized(
        const StringView &value,
        HashStringAllocator *allocator) {
        auto stream = initStream(allocator);

        const auto originalSize = stream.size();
        stream.appendStringView(value);
        ++size_;

        auto startAndFinish = allocator->finishWrite(stream, 1024);
        currentPosition_ = startAndFinish.second;
        POLLUX_CHECK_EQ(stream.size() - originalSize, value.size());
        return {startAndFinish.first};
    }

    namespace {
        HashStringAllocator::InputStream prepareRead(
            const AddressableNonNullValueList::Entry &entry) {
            auto header = entry.offset.header;
            auto seek = entry.offset.position - header->begin();

            HashStringAllocator::InputStream stream(header);
            stream.seekp(seek);
            return stream;
        }
    } // namespace

    // static
    bool AddressableNonNullValueList::equalTo(
        const Entry &left,
        const Entry &right,
        const TypePtr &type) {
        if (left.hash != right.hash) {
            return false;
        }

        auto leftStream = prepareRead(left);
        auto rightStream = prepareRead(right);

        CompareFlags compareFlags =
                CompareFlags::equality(CompareFlags::NullHandlingMode::kNullAsValue);
        return exec::ContainerRowSerde::compare(
                   leftStream, rightStream, type.get(), compareFlags) == 0;
    }

    // static
    void AddressableNonNullValueList::read(
        const Entry &position,
        BaseVector &result,
        vector_size_t index) {
        auto stream = prepareRead(position);
        exec::ContainerRowSerde::deserialize(stream, index, &result);
    }

    // static
    void AddressableNonNullValueList::readSerialized(
        const Entry &position,
        char *dest) {
        auto stream = prepareRead(position);
        stream.ByteInputStream::readBytes(dest, position.size);
    }
} // namespace kumo::pollux::aggregate::prestosql
